package com.etlmaster.executor.steps.kafka;

import java.util.ArrayList;
import java.util.List;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import com.etlmaster.executor.Step;
import com.etlmaster.executor.StepMeta;
import com.etlmaster.executor.annotation.StepInstruction;
import com.etlmaster.executor.realtime.kafka.KafkaHelper;
import com.etlmaster.executor.utils.StepState;

/**
 * @author Larry
 * @copyright Larry@etlMaster
 * @since 2017-09-10
 */
@StepInstruction("fetchMessage")
public class FetchMessFromKafkaStep extends Step {

	private FetchMessFromKafkaStepMeta meta;

	public FetchMessFromKafkaStep(StepMeta meta) {
		super(meta);
		this.meta = (FetchMessFromKafkaStepMeta) meta;
	}

	@Override
	protected int handle() {
		String topic = meta.getTopic();
		KafkaConsumer consumer;
		try {
			consumer = KafkaHelper.getConsumer(topic);

			ConsumerRecords<String, String> records = consumer.poll(10);
			List messageList = new ArrayList();
			if (records.count() > 0) {
				for (ConsumerRecord<String, String> record : records) {
					messageList.add(record.value());
					
				}
				meta.log("INFO", "从Kafka获取到主题{},下{}条消息", topic,
						messageList.size() + "");
				meta.getStepState().getReturnValue()
						.put("messageList", messageList);
			} else {
				meta.log("warn", "从Kafka获取到主题{},下未获取到消息", topic);
				return StepState.RUNNING_FAILED;
			}

			consumer.commitSync();
		} catch (Exception e) {
			meta.log("ERROR", e.getMessage());
		}
		return StepState.RUNNING_SUCCESSED;
	}

	@Override
	public boolean stop() {
		// TODO Auto-generated method stub
		return false;
	}

}
